
#pragma once
#include <muduo/net/TcpServer.h>
#include <muduo/net/TcpClient.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/Buffer.h>
#include <muduo/net/EventLoopThread.h>
#include <unordered_map>
#include <thread>
#include <chrono>
#include "detail.hpp"
#include "abstract.hpp"
#include "fields.hpp"
#include "message.hpp"

namespace myrpc{
    // 由基础缓冲区基类派生出muduo缓冲区子类 相关功能muduo库已实现 直接封装接口即可
    class MuduoBuffer : public BaseBuffer {
    public:
        using ptr = std::shared_ptr<MuduoBuffer>;
        MuduoBuffer(muduo::net::Buffer *buffer)
        :_buffer(buffer)
        {}
        virtual size_t readableBytes() override {
            return _buffer->readableBytes();
        }
        virtual int32_t peekInt32() override {
            return _buffer->peekInt32();  
        }
        virtual void retrieveInt32() override {
            _buffer->retrieveInt32();
        }
        virtual int32_t readInt32() override {
            return _buffer->readInt32();
        }
        virtual std::string retrieveAsString(size_t len) override {
            return _buffer->retrieveAsString(len);
        }

    private:
        muduo::net::Buffer *_buffer;
    };
    // 通过工厂模式生产指定类型的缓冲区对象
    class BufferFactory{
    public:
        template<typename BufferType, typename... Args>
        static BaseBuffer::ptr create(Args &&... args){
            return std::make_shared<BufferType>(std::forward<Args>(args)...);
        }
    };

    // 由基础协议基类派生出LV格式协议子类
    // |--total_len--|--VALUE--|
    // |--total_len--|--mtype--|--idlen--|--id--|--body--|
    class LVProtocal : public BaseProtocol {
    public:
        using ptr = std::shared_ptr<LVProtocal>;

        virtual bool canProcessed(const BaseBuffer::ptr &buf) override {
            if (buf->readableBytes() < lenFieldslength){
                return false;
            }
            int32_t total_len = buf->peekInt32();
            if (buf->readableBytes() < (total_len + lenFieldslength)){
                return false;
            }
            return true;
        }
        virtual bool onMessage(const BaseBuffer::ptr &buf, BaseMessage::ptr &msg) override {
            // 消息处理时默认视为缓冲区至少存在一条可处理的完整消息
            // 1.获取各个字段的长度
            int32_t total_len = buf->readInt32();
            MsgType mtype = static_cast<MsgType>(buf->readInt32());
            int32_t idlen = buf->readInt32();
            if (idlen < 0 || total_len < (mtypeFieldslength + idlenFieldslength + idlen)){
                root_logger->ERROR("收到无效的消息长度字段! total_len=%d, idlen=%d", total_len, idlen);
                return false;
            }
            int32_t body_len = total_len - idlen - idlenFieldslength - mtypeFieldslength;
            // 2.根据长度获取ID和消息主体
            std::string id = buf->retrieveAsString(idlen);
            std::string body = buf->retrieveAsString(body_len);
            // 3.由消息类型构建消息对象并完成消息主体的反序列化操作
            msg = MessageFactory::create(mtype);
            if (msg == nullptr){
                root_logger->ERROR("无法创建未知的消息类型: %d!", static_cast<int>(mtype));
                return false;
            }
            bool ret = msg->unserialize(body);
            if (ret == false){
                root_logger->ERROR("消息体反序列化(unserialize)失败!");
                return false;
            }
            // 4.设置消息对象类型以及ID
            msg->setMsgType(mtype);
            msg->setMsgId(id);
            return true;
        }
        virtual std::string serialize(const BaseMessage::ptr &msg) override {
            // 1.消息对象序列化获取消息主体 以及获取ID和消息类型
            std::string body = msg->serialize();
            std::string id = msg->getMsgId();
            MsgType h_mtype = msg->getMsgType();

            // 2.将字段从主机字节序转换为网络字节序
            int32_t n_mtype = htonl(static_cast<int32_t>(h_mtype));
            int32_t n_idlen = htonl(id.size());
            int32_t h_total_len = mtypeFieldslength + idlenFieldslength + id.size() + body.size();
            int32_t n_total_len = htonl(h_total_len);

            // 构建整个消息的序列化返回结果
            // |--total_len--|--mtype--|--idlen--|--id--|--body--|
            std::string result;
            result.reserve(h_total_len);
            result.append(reinterpret_cast<const char *>(&n_total_len), lenFieldslength);
            result.append(reinterpret_cast<const char *>(&n_mtype), mtypeFieldslength);
            result.append(reinterpret_cast<const char *>(&n_idlen), idlenFieldslength);
            result.append(id);
            result.append(body);
            return result;
        }

    private:
        const size_t lenFieldslength = 4;
        const size_t mtypeFieldslength = 4;
        const size_t idlenFieldslength = 4;
    };
    // 通过工厂模式生产指定类型的协议对象
    class ProtocalFactory{
    public:
        template<typename ProtocalType, typename... Args>
        static BaseProtocol::ptr create(Args &&... args){
            return std::make_shared<ProtocalType>(std::forward<Args>(args)...);
        }
    };

    // 由基础连接基类派生成Muduo形式连接的子类
    class MuduoConnection : public BaseConnection {
    public:
        using ptr = std::shared_ptr<MuduoConnection>;
        MuduoConnection(const BaseProtocol::ptr &protocal, const muduo::net::TcpConnectionPtr &conn)
        :_protocal(protocal),
        _conn(conn)
        {}

        virtual void send(const BaseMessage::ptr &msg) override {
            std::string str_msg = _protocal->serialize(msg);
            _conn->send(str_msg);
        }
        virtual void shutdown() override {
            _conn->shutdown();
        }
        virtual bool connected() override {
            return _conn->connected();
        }

    private: 
        BaseProtocol::ptr _protocal;
        muduo::net::TcpConnectionPtr _conn;
    };
    // 通过工厂模式生产指定类型的连接对象
    class ConnectionFactory{
    public:
        template<typename ConnectionType, typename... Args>
        static BaseConnection::ptr create(Args &&... args){
            return std::make_shared<ConnectionType>(std::forward<Args>(args)...);
        }
    };

    // 由基础服务端基类派生出Muduo服务端子类
    class MuduoServer : public BaseServer {
    public:
        using ptr=std::shared_ptr<MuduoServer>;
        MuduoServer(const int &port)
        :_server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "MuduoServer", muduo::net::TcpServer::kReusePort),
        _protocal(ProtocalFactory::create<LVProtocal>())
        {}

        void start(){
            // 设置系统层的回调处理 先执行系统级别的回调处理 然后再最后执行用户层的回调处理
            _server.setConnectionCallback(std::bind(&MuduoServer::onConnection, this, std::placeholders::_1));
            _server.setMessageCallback(std::bind(&MuduoServer::onMessage, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _server.start();
            _baseloop.loop();
        }

    private:
        // 连接建立和关闭的回调处理
        void onConnection(const muduo::net::TcpConnectionPtr &conn){
            // 连接建立 将连接封装成新的连接对象 并建立映射存放到集合中
            if (conn->connected()){
                server_logger->INFO("连接已建立!");
                auto muduo_conn = ConnectionFactory::create<MuduoConnection>(_protocal, conn);
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _conns.insert(std::make_pair(conn, muduo_conn));
                }
                if (_cb_conn){
                    _cb_conn(muduo_conn);
                }  
            }
            // 连接关闭 从连接集合中删除对应的映射关系
            else{
                server_logger->INFO("连接已关闭!");
                BaseConnection::ptr muduo_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if (it == _conns.end()){
                        return;
                    }
                    muduo_conn = it->second;
                    _conns.erase(conn);
                }
                if (_cb_close){
                    _cb_close(muduo_conn);
                }
            }
        }
        // 收到消息后的回调处理
        void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp){
            server_logger->DEBUG("连接有数据到来, 开始处理!");
            // 将缓冲区封装成新的缓冲区对象
            auto base_buf=BufferFactory::create<MuduoBuffer>(buf);
            while(true){
                if (_protocal->canProcessed(base_buf) == false){
                    server_logger->DEBUG("缓冲区的数据不足!");
                    break;
                }
                if (base_buf->readableBytes() > maxData){
                    conn->shutdown();
                    server_logger->WARNING("接收缓冲区数据过大(>%d bytes)，主动断开连接!", maxData);
                    return;
                }
                // 处理缓冲区中的数据获取消息对象
                BaseMessage::ptr msg;
                bool ret=_protocal->onMessage(base_buf, msg);
                if (ret == false){
                    conn->shutdown();
                    server_logger->WARNING("协议解析消息失败，主动断开连接!");
                    return;
                }
                // 根据连接获取封装后的连接对象
                BaseConnection::ptr base_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if (it == _conns.end()){
                        conn->shutdown();
                        server_logger->ERROR("出现内部错误：在连接集合中未找到当前连接!");
                        return;
                    }
                    base_conn = it->second;
                }
                // 调用消息回调函数进行处理
                if (_cb_message){
                    _cb_message(base_conn, msg);
                }
            }
        }
  
    private:
        const size_t maxData = (1 << 16);
        BaseProtocol::ptr _protocal;        // 服务端中收到的所有连接都共享同一个协议对象
        std::mutex _mutex;
        muduo::net::EventLoop _baseloop;
        muduo::net::TcpServer _server;
        std::unordered_map<muduo::net::TcpConnectionPtr, BaseConnection::ptr> _conns;
    };
    // 通过工厂模式生成指定类型的服务端对象
    class ServerFactory{
    public:
        template<typename ServerType, typename... Args>
        static BaseServer::ptr create(Args &&... args){
            return std::make_shared<ServerType>(std::forward<Args>(args)...);
        }
    };

    // 由基础客户端基类派生出Muduo客户端子类
    class MuduoClient : public BaseClient {
    public:
        using ptr=std::shared_ptr<MuduoClient>;
        MuduoClient(const std::string &ip, const int &port)
        :_protocal(ProtocalFactory::create<LVProtocal>()),
        _baseloop(_loopthread.startLoop()),
        _downlatch(1),
        _client(_baseloop, muduo::net::InetAddress(ip, port), "Muduoclient")
        {}

        virtual ~MuduoClient() {
            try {
                // 关键：先断开网络连接，再清理资源
                _client.disconnect();
                
                // 等待一小段时间，让网络线程完成清理
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
                
                // 再清理内部连接
                std::unique_lock<std::mutex> lock(_conn_mutex);
                if (_base_conn) {
                    _base_conn.reset();
                }
                // EventLoopThread会在_client析构时自动停止
            } catch (...) {
                // 析构时绝不能抛异常
            }
        }

        virtual void connect() override {
            try {
                _client.setConnectionCallback(std::bind(&MuduoClient::onConnection, this, std::placeholders::_1));
                _client.setMessageCallback(std::bind(&MuduoClient::onMessage, this, std::placeholders::_1,
                    std::placeholders::_2, std::placeholders::_3));
                _client.connect();
                _downlatch.wait();
            } catch (...) {
                client_logger->ERROR("连接建立异常");
                _downlatch.countDown();
            }
        }
        virtual void shutdown() override {
            try {
                if (_base_conn) {
                    _base_conn->shutdown();
                }
                _client.disconnect();
            } catch (...) {
                client_logger->ERROR("连接关闭异常");
            }
        }
        virtual bool connected() override {
            try {
                std::unique_lock<std::mutex> lock(_conn_mutex);
                return _base_conn && _base_conn->connected();
            } catch (...) {
                client_logger->ERROR("检查连接状态异常");
                return false;
            }
        }
        virtual bool send(const BaseMessage::ptr &msg) override {
            try {
                std::unique_lock<std::mutex> lock(_conn_mutex);
                if (_base_conn == nullptr){
                    client_logger->ERROR("客户端请求发送失败: 连接未建立或已断开!");
                    return false;
                }
                auto conn = _base_conn; // 获取连接副本
                lock.unlock();
                if (conn && conn->connected()) {
                    conn->send(msg);
                    return true;
                }
                return false;
            } catch (...) {
                client_logger->ERROR("消息发送异常");
                return false;
            }
        }
        virtual BaseConnection::ptr connection() override {
            std::unique_lock<std::mutex> lock(_conn_mutex);
            if (_base_conn && _base_conn->connected()){
                return _base_conn;
            }
            return BaseConnection::ptr();
        }

    private:
        void onConnection(const muduo::net::TcpConnectionPtr &conn){
            try {
                std::unique_lock<std::mutex> lock(_conn_mutex);
                if (conn->connected()){
                    client_logger->INFO("连接已建立!");
                    _base_conn = ConnectionFactory::create<MuduoConnection>(_protocal, conn);
                    lock.unlock(); 
                    _downlatch.countDown();
                    if (_cb_conn){
                        _cb_conn(_base_conn);
                    }
                }
                else{
                    client_logger->INFO("连接已关闭!");
                    _base_conn.reset();
                    lock.unlock(); 
                    if (_cb_close){
                        _cb_close(_base_conn);
                    }
                }
            } catch (...) {
                client_logger->ERROR("连接回调处理异常");
                _downlatch.countDown();
            }
        }
        void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp){
            try {
                auto base_buf = BufferFactory::create<MuduoBuffer>(buf);
                while(true){   
                    if (_protocal->canProcessed(base_buf) == false){
                        client_logger->DEBUG("缓冲区的数据不足!");
                        break;
                    }
                    if (base_buf->readableBytes() > maxDataSize){
                        client_logger->WARNING("接收缓冲区数据过大(>%d bytes)，连接将被断开!", maxDataSize);
                        conn->shutdown();
                        return;
                    }
                    BaseMessage::ptr msg;
                    bool ret = _protocal->onMessage(base_buf, msg);
                    if (ret == false){
                        client_logger->WARNING("协议解析消息失败，连接将被断开!");
                        conn->shutdown();
                        return;
                    }
                    if (_cb_message){
                        std::unique_lock<std::mutex> lock(_conn_mutex);
                        auto base_conn = _base_conn; // 获取连接的副本
                        lock.unlock();
                        if (base_conn) {
                            _cb_message(base_conn, msg);
                        }
                    }
                }
            } catch (...) {
                client_logger->ERROR("消息处理异常");
                try {
                    conn->shutdown();
                } catch (...) {
                    // 吞掉shutdown的异常
                }
            }
        }

    private:
        std::mutex _conn_mutex;
        const size_t maxDataSize = (1 << 16);
        BaseProtocol::ptr _protocal;
        BaseConnection::ptr _base_conn;
        muduo::CountDownLatch _downlatch;
        muduo::net::EventLoopThread _loopthread;
        muduo::net::EventLoop *_baseloop;
        muduo::net::TcpClient _client; 
    };
    // 通过工厂模式生成指定类型的客户端对象
    class ClientFactory{
    public:
        template<typename ClientType, typename... Args>
        static BaseClient::ptr create(Args &&... args){
            return std::make_shared<ClientType>(std::forward<Args>(args)...);
        }
    };
}